"""
Case Type   : Logical_Replication
Case Name   : 非初始用户使用-w创建、开启和删除逻辑复制槽失败，合理报错
Create At   : 2024/1/5
Owner       : xiaqi14
Description :
    1.修改参数
    2.创建sysadmin用户，配置用户白名单
    3.非初始用户不使用-w参数创建逻辑复制槽
    4.非初始用户使用-w参数创建逻辑复制槽
    5.非初始用户使用-w参数开启逻辑复制槽
    6.非初始用户使用-w参数删除逻辑复制槽
    7.非初始用户不使用-w参数删除逻辑复制槽
    8.删除用户
    9.恢复参数，恢复pg_hba.conf文件
Expect      :
    1.修改参数成功
    2.创建用户成功，配置白名单成功
    3.创建成功
    4.创建失败，合理报错
    5.开启失败，合理报错
    6.删除失败，合理报错
    7.删除成功
    8.删除成功
    9.恢复成功
History     :
"""
import os
import unittest
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant


class LogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.sh_primary = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.user_name = 'u_logical_replication_0067'
        self.slot_name_1 = 'slot_logical_replication_0067_1'
        self.slot_name_2 = 'slot_logical_replication_0067_2'
        self.config_wal_level = 'wal_level'
        self.wal_level_default = self.common.show_param(self.config_wal_level)

    def test_logical_replication(self):
        text = "-----step1:修改参数;expect:修改参数成功-----"
        self.logger.info(text)
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primary.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.common.show_param(self.config_wal_level)
        self.assertEquals(show_msg, 'logical', '执行失败' + text)

        text = "-----step2:创建sysadmin用户，配置用户白名单，" \
               "expect:创建用户成功，配置白名单成功-----"
        self.logger.info(text)
        sql_cmd = f'''drop user if exists {self.user_name};
            create user {self.user_name} sysadmin 
            password '{macro.COMMON_PASSWD}';'''
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG and
                      self.constant.CREATE_ROLE_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)
        param_1 = f'local replication {self.user_name}              sha256'
        param_2 = f'host  replication {self.user_name} 127.0.0.1/32 sha256'
        param_3 = f'host  replication {self.user_name} ::1/128      sha256'
        result_1 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step3:非初始用户不使用-w参数创建逻辑复制槽;expect:创建成功-----"
        self.logger.info(text)
        create_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--create', 'postgres', self.slot_name_1, self.user_name, '-s 2 ')
        self.logger.info(create_slot_msg)
        text = "-----验证创建逻辑复制槽成功;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name_1}%';"
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1, sql_msg, '执行失败:' + text)

        text = "-----step4:非初始用户使用-w参数创建逻辑复制槽;" \
               "expect:创建失败，合理报错-----"
        self.logger.info(text)
        create_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--create', 'postgres', self.slot_name_2, self.user_name, '-s 2 -w')
        self.logger.info(create_slot_msg)
        self.assertIn("FATAL:  Invalid username/password,login denied",
                      create_slot_msg, '执行失败:' + text)

        text = "-----step5:非初始用户使用-w参数开启逻辑复制槽;" \
               "expect:开启失败，合理报错-----"
        self.logger.info(text)
        start_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--start', 'postgres', self.slot_name_1, self.user_name,
            '-f - -o skip-empty-xacts=true -F 1 -s 2 -w')
        self.logger.info(start_slot_msg)
        self.assertIn("FATAL:  Invalid username/password,login denied",
                      start_slot_msg, '执行失败:' + text)

        text = "-----step6:非初始用户使用-w参数删除逻辑复制槽;" \
               "expect:删除失败，合理报错-----"
        self.logger.info(text)
        drop_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--drop', 'postgres', self.slot_name_1, self.user_name, '-s 2 -w')
        self.logger.info(drop_slot_msg)
        self.assertIn("FATAL:  Invalid username/password,login denied",
                      drop_slot_msg, '执行失败:' + text)

    def tearDown(self):
        text = "-----step7:非初始用户不使用-w参数删除逻辑复制槽;" \
               "expect:删除成功-----"
        self.logger.info(text)
        drop_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--drop', 'postgres', self.slot_name_1, self.user_name, '-s 2 ')
        self.logger.info(drop_slot_msg)
        sql_cmd = f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name_1}%';"
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)

        text_1 = "-----step8:删除用户;expect:删除成功-----"
        self.logger.info(text_1)
        drop_cmd = f'''drop user {self.user_name};'''
        drop_msg = self.sh_primary.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)

        text_2 = "-----step9:恢复参数，恢复pg_hba.conf文件，expect:恢复成功-----"
        self.logger.info(text_2)
        param_1 = f'local replication {self.user_name} '
        param_2 = f'host  replication {self.user_name} 127.0.0.1/32 '
        param_3 = f'host  replication {self.user_name} ::1/128 '
        result_1 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)

        wal_level_current = self.common.show_param(self.config_wal_level)
        if wal_level_current != self.wal_level_default:
            self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}={self.wal_level_default}')
            restart_msg = self.sh_primary.restart_db_cluster()
            self.logger.info(restart_msg)
        show_wal_level = self.common.show_param(self.config_wal_level)

        self.assertIn("(0 rows)", sql_msg, '执行失败:' + text)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_msg, '执行失败' + text_1)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text_2)
        self.assertEquals(show_wal_level, self.wal_level_default,
                          '执行失败' + text_2)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
